package com.xhwl.centiotdata.handler.eesd.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.xhwl.centiotdata.handler.eesd.EesdHandler;
import com.xhwl.centiotdata.mq.producer.GeneralProducer;
import com.xhwl.centiotdata.service.cent.IBiEnvCentService;
import com.xhwl.centiotdata.service.cent.IDeviceCentService;
import com.xhwl.centiotdata.service.cent.IProjectCentService;
import com.xhwl.centiotdata.utils.ThreadPoolExecutorService;
import com.xhwl.common.constant.cent.IOTReceiveConstant;
import com.xhwl.common.dto.cent.alarm.iot.IotAlarmBody;
import com.xhwl.common.dto.cent.device.DeviceDTO;
import com.xhwl.common.dto.cent.device.iot.IotDeviceJson;
import com.xhwl.common.dto.cent.device.iot.NotifyData;
import com.xhwl.common.dto.cent.eesd.IotAirVO;
import com.xhwl.common.dto.cent.eesd.IotEnvVO;
import com.xhwl.common.pojo.cent.bi.DwdFactEnvironmentMetrics;
import com.xhwl.common.pojo.cent.sdata.Organization;
import com.xhwl.common.query.cent.device.DevicePageQuery;
import com.xhwl.common.query.cent.sdata.OrganizationQuery;
import com.xhwl.common.utils.ChkUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import java.lang.reflect.Field;
import java.time.LocalDateTime;
import java.util.Arrays;
import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Author： lianghz
 * @Description：
 * @Date 2021/09/06/ 11:31
 */
@Service
@Slf4j
public class EesdHuaWeiBaHandler extends EesdHandler {

    @Autowired
    IDeviceCentService deviceCentService;

    @Autowired
    IBiEnvCentService iBiEnvCentService;

    @Autowired
    private GeneralProducer generalProducer;

    @Autowired
    IProjectCentService iProjectCentService;


    public static Map<String, DeviceDTO> deviceMap = new ConcurrentHashMap<>(4096);

    public static Map<String, String> treeMap = new ConcurrentHashMap<>(4096);

    @Override
    public void afterPropertiesSet() {

    }

    private final static String HUA_WEI_BA_TOPIC = "HUA_WEI_BA_TOPIC";

    /**
     * 消息入kafka队列
     */
    public void pullBaRecord(String t) {
        IotDeviceJson iotDeviceJson = JSONObject.parseObject(t, IotDeviceJson.class);
        generalProducer.send(HUA_WEI_BA_TOPIC, iotDeviceJson);
    }

    /**
     * 消息出kafka队列
     */
    @KafkaListener(topics = HUA_WEI_BA_TOPIC, groupId = HUA_WEI_BA_TOPIC)
    public void getQueue(ConsumerRecord record) {
        if (record.value() == null || record.value().toString().length() == 0) {
            return;
        }
        ThreadPoolExecutorService.getThreadPool().execute(() -> {
            try {
                String val = record.value().toString();
                handleMsg(val);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }

    /**
     * 处理消息
     */
    private void handleMsg(String t) {
        Map<String, Object> res = check(t);
        List<IotAlarmBody> services = (List<IotAlarmBody>) res.get("services");
        String deviceId = (String) res.get("deviceId");

        List<DeviceDTO> devices = Lists.newArrayList();
        List<DwdFactEnvironmentMetrics> pm25s = Lists.newArrayList();
        List<DwdFactEnvironmentMetrics> temps = Lists.newArrayList();

        DeviceDTO deviceDTO;
        for (IotAlarmBody iotAlarmBody : services) {
            DevicePageQuery devicePageQuery = new DevicePageQuery();
            devicePageQuery.setIotDeviceId(deviceId);
//            log.error("设备ID:{}", deviceId);
            if (!deviceMap.containsKey(deviceId)) {
                Page<DeviceDTO> devicePage = deviceCentService.pageDevice(null, devicePageQuery);
                if (ChkUtil.isNull(devicePage) || ChkUtil.isNull(devicePage.getRecords())) {
                    deviceMap.put(deviceId, new DeviceDTO());
                    continue;
                }
                //处理PM2.5
                deviceDTO = devicePage.getRecords().get(0);
                deviceMap.put(deviceDTO.getIotDeviceId(), deviceDTO);
            } else {
                deviceDTO = deviceMap.get(deviceId);
            }
            if (ChkUtil.isNull(deviceDTO) || ChkUtil.isNull(deviceDTO.getId())) {
//                log.error("====================>设备中台返回的设备为空！！！！！");
                continue;
            }
            if (iotAlarmBody.getService_id().equals(IOTReceiveConstant.AIR_QUALITY)) {
                try {
                    pm25s.add(pullPm25Record(iotAlarmBody, deviceDTO));
                } catch (Exception ignore) {
                }
            }
            //处理温湿度 FreshAirHandlingUnit[新风机]
            //之前是iotAlarmBody.getService_id().equals(IOTReceiveConstant.AIR_CONDITIONER_DEVICE)
            //进行修改
            if ("FreshAirHandlingUnit".equals(iotAlarmBody.getService_id())) {
                try {
                    List<DwdFactEnvironmentMetrics> dwds = pullTempRecord(iotAlarmBody, deviceDTO);
                    temps.addAll(dwds);
                } catch (Exception ignore) {
                }
            }
            //处理设备属性
            try {
                devices.add(attr(iotAlarmBody, deviceDTO));
            } catch (Exception ignore) {
            }
        }
        if (ChkUtil.isNotNull(devices)) {
            deviceCentService.batchUpdate(null, devices);
        }
        if (ChkUtil.isNotNull(pm25s)) {
            iBiEnvCentService.asyncEnvInitData(JSON.toJSONString(pm25s));
        }
        if (ChkUtil.isNotNull(temps)) {
            iBiEnvCentService.asyncEnvInitData(JSON.toJSONString(temps));
        }
    }

    /**
     * 处理设备属性
     */
    private DeviceDTO attr(IotAlarmBody iotAlarmBody, DeviceDTO deviceDTO) {
        DeviceDTO device = new DeviceDTO();
        device.setId(deviceDTO.getId());
        JSONObject jsonObject = JSONObject.parseObject(JSONObject.toJSONString(iotAlarmBody.getProperties()));
        jsonObject.put("deviceServiceId", iotAlarmBody.getService_id());
        jsonObject.put("eventTime", iotAlarmBody.getEvent_time().toString());
        device.setAttributeJson(jsonObject);
        return device;
    }

    /**
     * 拉取温度 AirQuality
     */
    private List<DwdFactEnvironmentMetrics> pullTempRecord(IotAlarmBody iotAlarmBody, DeviceDTO deviceDTO) throws Exception {
        IotAirVO properties = JSONObject.parseObject(JSONObject.toJSONString(iotAlarmBody.getProperties()), IotAirVO.class);
        Class<? extends IotAirVO> aClass = properties.getClass();
        List<DwdFactEnvironmentMetrics> list = Lists.newArrayList();
        String orgId = ChkUtil.getString(deviceDTO.getOrganizationId());
        LocalDateTime eventTime = iotAlarmBody.getEvent_time();
        Integer projectId = deviceDTO.getProjectId();
        DwdFactEnvironmentMetrics values;
        Set<Double> excludeKeys = new HashSet<>();
        excludeKeys.add(-0.1d);
        excludeKeys.add(0.1d);
        excludeKeys.add(0d);
        for (int i = 1; i < 12; i++) {
            try {
                //温度
                values = getBaseInfo(projectId, orgId, deviceDTO, eventTime);
                initData(i, values, "T", aClass, properties, list, excludeKeys);
                //湿度
                values = getBaseInfo(projectId, orgId, deviceDTO, eventTime);
                initData(i, values, "H", aClass, properties, list, excludeKeys);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return list;
    }

    /**
     * {"resource":"device.property",
     * "notify_data":{"header":{"device_id":"61987f96d50ad802887831e5_1641610521258","product_id":"61987f96d50ad802887831e5",
     * "app_id":"c9c3f09f28434760b4f2a0b6c4f116a2","gateway_id":"61987f96d50ad802887831e5_1641610521258","node_id":"1641610521258"},
     * "body":{"services":[{"service_id":"AirQuality","properties":{"PM25":"2000","CO2":"2000","PM10":"2000"},"event_time":"20220108T031949Z"}]}},
     * "event":"report","event_time":"20220108T031949Z"}
     * 拉取PM2.5 AirQuality
     */
    private DwdFactEnvironmentMetrics pullPm25Record(IotAlarmBody iotAlarmBody, DeviceDTO deviceDTO) {
        IotEnvVO properties = JSONObject.parseObject(JSONObject.toJSONString(iotAlarmBody.getProperties()), IotEnvVO.class);
        String orgId = ChkUtil.getString(deviceDTO.getOrganizationId());
        LocalDateTime eventTime = iotAlarmBody.getEvent_time();
        Integer projectId = deviceDTO.getProjectId();
        DwdFactEnvironmentMetrics values = getBaseInfo(projectId, orgId, deviceDTO, eventTime);
        values.setType("pm2p5");
        values.setDeviceCode(deviceDTO.getIotDeviceId());
        values.setPm10(Double.valueOf(properties.getPM10()));
        values.setPm2p5(Double.valueOf(properties.getPM25()));
        values.setCo2(Double.valueOf(properties.getCO2()));
        return values;
    }

    /**
     * 获取基本信息
     */
    public DwdFactEnvironmentMetrics getBaseInfo(Integer projectId, String orgId, DeviceDTO deviceDTO, LocalDateTime eventTime) {
        DwdFactEnvironmentMetrics values = new DwdFactEnvironmentMetrics();
        values.setProjectId(projectId);
        values.setRoomId(ChkUtil.getString(getOrgParentId(orgId), "0"));
        values.setFloorId(ChkUtil.getString(orgId, "0"));
        values.setProductId(deviceDTO.getId());
        values.setDeviceId(deviceDTO.getIotDeviceId());
        values.setGroupCode("教室");
        values.setName(deviceDTO.getName());
        values.setUpdateTime(eventTime);
        values.setCreateTime(eventTime);
        return values;
    }

    /**
     * 从儿子获取顶级楼层
     */
    private String getOrgParentId(String orgId) {
        //获取顶级宿舍楼A,宿舍楼B完整的路址ID
        if (ChkUtil.isNull(orgId)) {
            return orgId;
        }
        String parentId = treeMap.get(orgId);
        if (ChkUtil.isNotNull(parentId)) {
            return parentId;
        }
        //向上取宿舍楼BID {"roundLevel": 10, "id":223, "type":3}
        OrganizationQuery query = new OrganizationQuery();
        query.setId(ChkUtil.getInteger(orgId));
        query.setRoundLevel(10);
        query.setTypes(Arrays.asList(3));
        List<Organization> organizations = iProjectCentService.organizationList(query);
        Optional<Organization> min = organizations.stream().min(Comparator.comparing(Organization::getLevel));
        if (min.isPresent()) {
            Organization organization = min.get();
            parentId = ChkUtil.getString(organization.getId());
        }
        String string = ChkUtil.getString(parentId, orgId);
        treeMap.put(orgId, string);
        return string;
    }

    /**
     * 初始化温湿度
     */
    private void initData(int i, DwdFactEnvironmentMetrics values, String key, Class<? extends IotAirVO> aClass, IotAirVO properties, List<DwdFactEnvironmentMetrics> list, Set<Double> excludeKeys) throws Exception {
        String tempField = key + i;
        Field field = aClass.getDeclaredField(tempField);
        field.setAccessible(true);
        String value = ChkUtil.getString(field.get(properties));
        if (ChkUtil.isNull(value)) {
            return;
        }
        double val = Double.parseDouble(value);
        //一些特殊值过滤
        if (excludeKeys.contains(val)) return;
        values.setDeviceCode(values.getDeviceId() + "_" + tempField);
        if ("T".equals(key)) {
            values.setHumidity(null);
            values.setType("temp");
            values.setTemp(val);
        } else {
            values.setTemp(null);
            values.setType("humidity");
            values.setHumidity(val);
        }
        //添加
        list.add(values);
    }

    /**
     * 异常判断
     */
    private <T> Map<String, Object> check(String t) {
        IotDeviceJson iotDeviceJson = JSONObject.parseObject(t, IotDeviceJson.class);
        if (ChkUtil.isNull(iotDeviceJson)) {
            throw new RuntimeException("==================>iotDeviceJson消息异常！<===================");
        }
        NotifyData notifyData = iotDeviceJson.getNotify_data();
        if (ChkUtil.isNull(notifyData) || ChkUtil.isNull(notifyData.getBody()) || ChkUtil.isNull(notifyData.getHeader())) {
            throw new RuntimeException("==================>notifyData||getBody()消息异常！<===================");
        }
        String deviceId = notifyData.getHeader().getDevice_id();
        if (ChkUtil.isNull(deviceId)) {
            throw new RuntimeException("==================>deviceId消息异常！<===================");
        }
        JSONObject body = JSONObject.parseObject(notifyData.getBody().toString());
        List<IotAlarmBody> services = JSONArray.parseArray(body.get(IOTReceiveConstant.SERVICES).toString(), IotAlarmBody.class);
        if (ChkUtil.isNull(services)) {
            throw new RuntimeException("==================>services消息异常！<===================");
        }
        Map<String, Object> map = Maps.newHashMap();
        map.put("services", services);
        map.put("deviceId", deviceId);
        return map;
    }
}
